package com.shujia.stream

import org.apache.spark.streaming.dstream.{DStream, ReceiverInputDStream}
import org.apache.spark.{SparkConf, SparkContext}
import org.apache.spark.streaming.{Durations, StreamingContext}

object Demo2Window {
  def main(args: Array[String]): Unit = {
    /**
      * 1、创建SparkContext
      *
      */
    val conf: SparkConf = new SparkConf()
      .setMaster("local[2]")
      .setAppName("stream")

    val sc = new SparkContext(conf)

    /**
      * 2、创建SparkStreaming环境
      *
      * 指定batch的间隔时间
      */

    val ssc = new StreamingContext(sc, Durations.seconds(5))

    ssc.checkpoint("data/checkpoint")

    val linesDS: ReceiverInputDStream[String] = ssc.socketTextStream("master", 8888)

    val kvDS: DStream[(String, Int)] = linesDS.flatMap(_.split(",")).map((_, 1))

    /**
      * 每隔5秒统计最近15秒单词的数量
      *
      */

    /*    val cuntDS: DStream[(String, Int)] = kvDS.reduceByKeyAndWindow(
          (x: Int, y: Int) => x + y, //窗口内的聚合函数
          Durations.seconds(15), //窗口大小
          Durations.seconds(5) //滑动时间
        )*/


    //优化版本
    val cuntDS: DStream[(String, Int)] = kvDS.reduceByKeyAndWindow(
      (x: Int, y: Int) => x + y, //窗口内的聚合函数
      (i: Int, j: Int) => i - j, //减去多于数据的函数
      Durations.seconds(15), //窗口大小
      Durations.seconds(5) //滑动时间
    )

    //过滤掉为0的数据
    val filterDS: DStream[(String, Int)] = cuntDS.filter(_._2 != 0)

    filterDS.print()

    ssc.start()
    ssc.awaitTermination()
    ssc.stop()

  }

}
